package com.simple.datax.seda.core;
import java.util.ArrayList;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import com.simple.datax.seda.api.QueueElementIF;
import com.simple.datax.seda.api.StageWrapperIF;
import com.simple.datax.seda.api.ThreadManagerIF;
import com.simple.datax.seda.core.event.HeartbeatEvent;
import com.simple.datax.seda.core.event.NullEvent;
import com.simple.datax.seda.core.event.StageInterruptEvent;
import com.simple.datax.seda.core.exception.InitFailedException;
import com.simple.datax.seda.core.exception.UnknownStageEventException;

public class MultiThreadManager implements ThreadManagerIF {

	private Log LOG = LogFactory.getLog(MultiThreadManager.class);

	private final Object LOCKER = new Object();

	private StageWrapperIF stageWrapper;

	private int threadCount;

	private ThreadGroup threadGroup;

	private String name;

	private ArrayList<Thread> threadList = new ArrayList<Thread>();

	private InterruptController interruptController;

	public MultiThreadManager(String name) throws InitFailedException {
		if (name == null || name.trim().equals("")) {
			throw new InitFailedException("线程管理器[name]属性是空.");
		}
		this.stageWrapper = null;
		this.threadGroup = new ThreadGroup("MultiThreadManager:" + name);
		this.threadCount = 1;
		this.name = name;
		this.threadList = new ArrayList<Thread>();
		interruptController = new InterruptController();
	}

	public boolean isAlive() {
		return (threadGroup.activeCount() > 0);

	}

	public void joinAll() {
		for (int i = 0; i < threadList.size(); i++) {
			try {
				threadList.get(i).join();
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		threadList.clear();
		threadList = null;
		threadGroup.destroy();
		threadGroup = null;
	}

	public void setStageWrapper(StageWrapperIF stageWrapper) {
		this.stageWrapper = stageWrapper;
	}

	public void initialize() throws InitFailedException {
		if (this.stageWrapper == null) {
			throw new InitFailedException("线程管理器 [" + name
					+ "] 的 stageWrapper 属性是空.");
		}
		LOG.info("线程管理器 [" + name + "] 初始化成功.");
	}

	// public void setConfig(ConfigInfoIF config) throws InitFailedException{
	// // try{
	// // int count = config.getInt(MultiThreadManager.COUNT_THREAD);
	// // if (count < 1){
	// // throw new InvalidPropertyValueException();
	// // }
	// // }catch(UnknownKeyNameException e){
	// // LOG.warn("Unknown key name ["+MultiThreadManager.COUNT_THREAD+ "].");
	// // }
	// LOG.info("线程管理器 ["+name+"] 初始化成功.");
	// }

	public void start() {
		for (int i = 0; i < threadCount; i++) {
			Thread thread = new Thread(threadGroup, new AppThread(name + "-"
					+ (i + 1)), name + "-" + (i + 1));
			thread.start();
			threadList.add(thread);
		}
		LOG.info("线程管理器 [" + name + "]启动成功,共启动[" + threadCount + "] 个线程.");
	}

	public boolean processOnce(String serialNumber) {
		QueueElementIF aEvent = this.stageWrapper.pollEvent();

		// 已经收到了准备中断线程管理器的命令
		if (interruptController.isInterruptting()) {
			if (aEvent == null) {
				// 如果接收到nullEvent,自己退出,并判断如果当前已经是最后一个活动县城了,就生成一个StageInterruptEvent
				// 交给EventHandler处理
				synchronized (LOCKER) {
					if (interruptController.getActiveCount() <= 1) {
						// 这里是通知其他线程管理器退出的唯一出口
						handleEvent(serialNumber, new StageInterruptEvent());
					}
					// 这里是线程中断的唯一出口
					interruptController.interrupt();
					return false;
				}

			} else {// 如果接收到的event不是null,则继续进行处理
				if (aEvent instanceof HeartbeatEvent) {
					HeartbeatEvent hbEvent = (HeartbeatEvent) aEvent;
					hbEvent.report();
					return true;
				} else if (aEvent instanceof StageInterruptEvent) {
					LOG
							.debug("线程 [" + serialNumber
									+ "] 收到一个中断命令, 线程进入准备中断状态.");
					interruptController.interrupting();
					return true;
				}

				handleEvent(serialNumber, aEvent);
				return true;
			}
		} else {
			if (aEvent == null) {
				handleEvent(serialNumber, new NullEvent());
				return true;
			} else {
				boolean success = true;
				if (aEvent instanceof HeartbeatEvent) {
					HeartbeatEvent hbEvent = (HeartbeatEvent) aEvent;
					hbEvent.report();
					return true;
				} else if (aEvent instanceof StageInterruptEvent) {
					LOG
							.debug("线程 [" + serialNumber
									+ "] 收到一个中断命令, 线程进入准备中断状态.");
					interruptController.interrupting();
					return true;
				}
				return handleEvent(serialNumber, aEvent);
			}

		}

		// if (aEvent == null) {
		// if (interruptController.isInterruptting()){
		// return handleEvent(serialNumber, new StageInterruptEvent());
		// }else{
		//				
		// }
		// }
		//			
		// boolean success = false;
		// long begin = System.currentTimeMillis();
		// if (aEvent instanceof HeartbeatEvent){
		// HeartbeatEvent hbEvent = (HeartbeatEvent)aEvent;
		// hbEvent.report();
		// return true;
		// }else if (aEvent instanceof StageInterruptEvent) {
		// LOG.info("Thread ["+serialNumber+"] get a StageInterruptEvent, the
		// thread is to be end.");
		// interruptController.interrupt();
		// return true;
		// }
		//		
		// return handleEvent(serialNumber, aEvent);

		// EventStatistic eventStatistic =
		// stageWrapper.getEventStatistic(aEvent.getClass().getName());
	}

	private boolean handleEvent(String serialNumber, QueueElementIF aEvent) {

		try {
			boolean isOk = this.stageWrapper.getEventHandler().handleEvent(
					aEvent);
			if (!isOk) {
				LOG.info("线程进入中断.");
				interruptController.interrupting();
			}
			return true;
		} catch (UnknownStageEventException e) {
			LOG.error("线程[" + serialNumber + "]接收到未知命令\n" + e.getClass() + ":"
					+ e.getMessage());
			return true;
		} catch (InitFailedException e) {
			LOG.fatal("线程[" + serialNumber + "]发生初始化异常\n" + e.getClass() + ":"
					+ e.getMessage());
			System.exit(-1);
		} catch (Exception e) {
			LOG.error("线程[" + serialNumber + "]发生异常\n" + e.getClass() + ":"
					+ e.getMessage());
			return true;
		} catch (Error e) {
			LOG.fatal("线程[" + serialNumber + "]发生错误,导致系统退出\n" + e.getClass()
					+ ":" + e.getMessage());
			System.exit(-1);
		} finally {
			long end = System.currentTimeMillis();
			// eventStatistic.addNewProcess(success, end-begin);
		}
		return true;
	}

	public int getThreadCount() {
		return threadCount;
	}

	public void setThreadCount(int threadCount) {
		this.threadCount = threadCount;
	}

	private class AppThread implements Runnable {

		private String serialNumber;

		AppThread(String sn) {

			this.serialNumber = sn;
		}

		public void run() {
			LOG.debug("线程 [" + serialNumber + "] 启动.");
			interruptController.start();
			while (true) {
				if (!processOnce(serialNumber)) {
					LOG.debug("线程 [" + serialNumber + "] 处理失败.");
					// stageWrapper.getEventHandler().clear();
					break;
				}
			}

			LOG.debug("线程 [" + serialNumber + "] 结束.");
		}
	}

}
